﻿#include <iostream>
#define SYNC_BOOST_SOCKET_TCP
#ifdef SYNC_BOOST_SOCKET_TCP
#include "ring_buffer.h"

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/shared_ptr.hpp>
#include "server.h"

#define DATA_LEN (1<<24) //4MB
#define CACHE_LEN (1 << 29) //512MB


#ifdef _WIN32
typedef struct
{
    uint8_t data[DATA_LEN - 4];
    unsigned int	count;
}pket;
#else
typedef struct
{
    uint8_t data[DATA_LEN - 4];
    unsigned int	count;
}__attribute__((packed)) pket;

#endif
// (__attribute__ package)pket;

using namespace std;
using namespace boost::asio;


bool sockValid = false;
ring_buffer rbuf;
ring_buffer sbuf;
// 所有asio类都需要io_service对象
io_service iosev;
ip::tcp::socket client(iosev);

bool aceptThrdExit = false;
bool txThrdExit = false;
bool rxThrdExit = false;
#ifdef OS_LINUX
pthread_t sid;
pthread_t rid;
pthread_t sigid;
#endif

void reconnect();

void* recvRoutin(void* handle)
{
    //if(!handle)
    //	return;
    //Server *p = (xServer*)handle;
    cout << "start Recv service.." << endl;
    boost::system::error_code ec;
    while (!rxThrdExit)
    {
        //char buf[12] = {0x48,49,0x50,0x51,0x52,0x53,0x54,0x55,0x56,0x57,0x58,0x59};
        if (sockValid)
        {
            uint8_t buf2[13];
            int len1 = read(client, buffer(buf2), transfer_exactly(13), ec);
#ifdef __DEBUF
            cout << "server recv  len = " << len1 << endl;
#endif
            if (len1 == 0)
            {
                sockValid = false;
                client.close();
                cout << __func__ << ":" << __LINE__ << "Link down!!!" << endl;
            }
            else
            {
                // int count = ring_buffer_put(&rbuf, buf2, len1);
#ifdef __DEBUG
                cout << "save to buf " << count << "byte:" << buf2 << endl;
#endif
            }
        }
        else
        {
            // cout << __func__ << ":" << __LINE__ << "Link down!!!" << endl;
#ifdef __OS_LINUX
            usleep(1000 * 1000);
#else
            Sleep(500);
#endif
        }
    }
    cout << "rxRoutin Exit normal!!" << endl;
    return NULL;
}


void* sendRoutin(void* handle)
{
    //if(!handle)
    //	return;
    //Server *p = (xServer*)handle;
    cout << "start Send service.." << endl;
    boost::system::error_code ec;
    while (!txThrdExit)
    {
        //char buf[12] = {0x48,49,0x50,0x51,0x52,0x53,0x54,0x55,0x56,0x57,0x58,0x59};
        if (sockValid)
        {
            if (ring_buffer_len(&sbuf) >= DATA_LEN)
            {
                static uint8_t buf1[DATA_LEN];
                ring_buffer_get(&sbuf, buf1, DATA_LEN);
                unsigned int len = ((pket*)buf1)->count;
                if (len == 0)
                    continue;

                int len1 = write(client, buffer(buf1), transfer_exactly(len), ec);
                //int len1 = client.write_some(buffer(buf1));
                if (len1 == 0)
                {
                    sockValid = false;
                    client.close();
                    cout << __func__ << ":" << __LINE__ << "Link down!!!" << endl;
                }
            }
        }
        else
        {
            // cout <<__func__<<":"<<__LINE__<< "Link down!!!" << endl;
#ifdef __OS_LINUX
            usleep(500);
#else
            Sleep(500);
#endif
        }
    }
    cout << "txRoutin Exit normal!!" << endl;
    return NULL;
}

#ifdef USE_IGNORE_SIGPIPE 
void signal_handler(const boost::system::error_code& err, int signal)
{
    switch (signal) {
    case SIGPIPE:
        std::cout << "SIGNINT" << std::endl;
        break;
    case SIGTERM:
        std::cout << "SIGNTERM" << std::endl;
        break;
    default:
        break;
    }
}

#endif

void* Accept(void* p)
{
    short port = *((short*)p);
    while (!aceptThrdExit)
    {
        if (!sockValid)
        {
            ip::tcp::acceptor acceptor(iosev,
                ip::tcp::endpoint(ip::tcp::v4(), port));
            acceptor.accept(client);
            sockValid = true;
            cout << __func__ << ":" << __LINE__ << "客户端连接成功!!!" << endl;
        }
#ifdef __OS_LINUX
        sleep(1);
#else
        Sleep(1000);
#endif
    }
    cout << "aceptThread Exit normal!!" << endl;

    return NULL;
}

int server::Send(uint8_t* buf, int len)
{
    if (len > DATA_LEN)
    {
        cout << "package data over 1MB!!!" << endl;
        return -1;
    }

    pket p;
    memcpy(p.data, buf, len);
    p.count = len;
    cout << __LINE__ << ": " << p.count << endl;
    return ring_buffer_put(&sbuf, (uint8_t*)&p, DATA_LEN);
}

int Recv(uint8_t* buf, int len)
{
    int len1 = ring_buffer_len(&rbuf);
#ifdef __DEBUG
    cout << "rbuf has " << len1 << "data" << endl;
#endif
    if (len1 >= len)
        return ring_buffer_get(&rbuf, buf, len);
    else
    {
        cout << "no data" << endl;
        return 0;
    }
}

boost::thread* tThrd = new boost::thread();
boost::thread* rThrd = new boost::thread();
boost::thread* aThrd = new boost::thread();


int initServer(short lport)
{
    //server socket prepare and waiting for connect
    aThrd = new boost::thread(&Accept, &lport);
    //memcpy(aThrd, &acptThrd, sizeof(boost::thread));

    //1. init socket sendbuf(4MB)
    char* buf = (char*)calloc(1, CACHE_LEN);
    if (buf)
        ring_buffer_init(&sbuf, buf, CACHE_LEN);
    else
        return -1;

    char* mem = (char*)calloc(1, CACHE_LEN);
    if (mem)
        ring_buffer_init(&rbuf, mem, CACHE_LEN);
    else
    {
        free(buf);
        return -1;
    }

    //2.start send engin
    tThrd = new boost::thread(&sendRoutin, &sbuf);
    //memcpy(tThrd, &txThrd, sizeof(boost::thread));
#ifdef OS_LINUX
    pthread_create(&sid, NULL, sendRoutin, &sbuf);
#endif
    //3.start recieve engin
    rThrd = new	boost::thread(&recvRoutin, &rbuf);
    //memcpy(rThrd, &rxThrd, sizeof(boost::thread));

    return 0;
}

#ifdef USE_IGNORE_SIGPIPE
void* registerSig(void* p)
{
    cout << "register sig pipe.." << endl;
    signal_set sigset(iosev, SIGPIPE, SIGTERM);
    sigset.async_wait(signal_handler);
    boost::system::error_code ec;
    iosev.run(ec);
}

#endif 


#endif
#ifdef __MAIN__
int lPort = 11002;
int main(int argc, char* argv[])
{
    int ret = initServer();
    if (ret < 0)
    {
        cout << "init server error, check..." << endl;
        return -1;
    }

    //send test data
    uint8_t buf1[12] = { 0x48,49,0x50,0x51,0x52,0x53,0x54,0x55,0x56,0x57,0x58,0x59 };

    uint8_t recvData[13];
    while (1)
    {
        if (sockValid)
        {
            int scount = Send(buf1, sizeof(buf1));
            //cout<<"server send count:"<<scount;
            Sleep(1000);

            int rcount = Recv(recvData, 13);
            if (rcount > 0)
                cout << recvData << endl;
        }
        else
            Accept(lPort);

    }

    return 0;
}
#endif

void destroyServer()
{
    aceptThrdExit = true;
    rxThrdExit = true;
    txThrdExit = true;
#ifdef __OS_LINUX
    sleep(2);
    #else
        Sleep(2000);
#endif
    //2.stop recvRoutin
    rThrd->join();
    delete  rThrd;
    //3.stop sendRoutin
    tThrd->join();
    delete tThrd;
    //4.release sendbuf
    free(rbuf.buffer);
    //5.release recvbuf
    free(sbuf.buffer);

    //1.stop acptThread
    aThrd->join();
    delete aThrd;

}
server::server(short lport)
{
    int ret = initServer(lport);
    if (ret < 0)
        cout << "init server error, check..." << endl;
}


server::~server()
{
    destroyServer();
}
